package org.apache.rocketmq.client.codecoord.producer;

import org.apache.rocketmq.client.codecoord.RocketMqConstant;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;

/**
 * 异步消息发送
 *
 * @author tianxincoord@163.com
 * @since 2021/10/25
 */
public class BasicAsyncMessageSend {

    public static void main(String[] args) throws Exception {
        ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "生产者");
            }
        });
        // 注意异步消息只能创建一个对象，不然会报组重复
        DefaultMQProducer producer = new DefaultMQProducer(RocketMqConstant.GROUP);
        producer.setNamesrvAddr(RocketMqConstant.NAME_SRV_ADDRESS);
        producer.start();
        // 指定异步发送失败后不进行重试发送
        producer.setRetryTimesWhenSendFailed(0);
        // 指定新创建的Topic的Queue数量为2，默认为4
        producer.setDefaultTopicQueueNums(2);

        BasicAsyncMessageSend messageSend = new BasicAsyncMessageSend();
        messageSend.asyncProducer(producer);
        /// 循环发送
        /*executor.scheduleWithFixedDelay(() -> {
            try {
                messageSend.asyncProducer(producer);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, 0, 1, TimeUnit.SECONDS);*/
    }

    /**
     * 发送异步消息
     */
    public void asyncProducer(DefaultMQProducer producer) throws Exception {
        long timeMillis = System.currentTimeMillis();
        String message = "异步消息 " + timeMillis;
        Message msg = new Message(RocketMqConstant.TOPIC, RocketMqConstant.TAG, message.getBytes(RemotingHelper.DEFAULT_CHARSET));
        msg.setKeys(String.valueOf(timeMillis));
        // 比同步多了SendCallback回调方法
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(String.format("消息【%s】发送成功【%s】", message, sendResult));
            }

            @Override
            public void onException(Throwable e) {
                System.err.println(String.format("消息【%s】发送异常【%s】", message, e.getMessage()));
            }
        });
    }
}
